package com.wzy.gmail.realtime.util

import com.alibaba.fastjson.JSON
import com.alibaba.fastjson.serializer.SerializeConfig
import org.apache.http.HttpHost
import org.elasticsearch.action.bulk.BulkRequest
import org.elasticsearch.action.index.IndexRequest
import org.elasticsearch.action.search.{SearchRequest, SearchResponse}
import org.elasticsearch.client.indices.GetIndexRequest
import org.elasticsearch.client.{RequestOptions, RestClient, RestClientBuilder, RestHighLevelClient}
import org.elasticsearch.common.xcontent.XContentType
import org.elasticsearch.search.SearchHit
import org.elasticsearch.search.builder.SearchSourceBuilder

import java.util
import scala.collection.Searching.SearchResult
import scala.collection.mutable.ListBuffer


/**
 * ES工具类，对ES进行读写操作
 */
object MyEsUtils {
  /**
   * 客户端对象
   */
  val esClient: RestHighLevelClient = build()

  /**
   * 创建客户端对象
   */
  def build(): RestHighLevelClient = {
    val host: String = MyPropsUtils(MyConfig.ES_HOST)
    val port: String = MyPropsUtils(MyConfig.ES_PORT)
    val clientBuilder: RestClientBuilder = RestClient.builder(new HttpHost(host, port.toInt))
    val client = new RestHighLevelClient(clientBuilder)
    client
  }

  /**
   * 关闭客户端对象
   */
  def close(): Unit = {
    if (esClient != null) esClient.close()
  }

  /**
   * 1.批量写入
   * 2.幂等写
   */
  def bulkSave(indexName: String, docs: List[(String, AnyRef)]): Unit = {
    val bulkRequest: BulkRequest = new BulkRequest()
    for ((docId, docObj) <- docs) {
      val indexRequest: IndexRequest = new IndexRequest(indexName)
      val dataJson: String = JSON.toJSONString(docObj, new SerializeConfig(true))
      indexRequest.source(dataJson, XContentType.JSON)
      bulkRequest.add(indexRequest)
    }
    esClient.bulk(bulkRequest, RequestOptions.DEFAULT)
  }
  def searchField(indexName:String,fieldName:String):List[String]={
    // 判断索引是否存在
    val indexRequest:GetIndexRequest = new GetIndexRequest(indexName)
    val isExists: Boolean = esClient.indices().exists(indexRequest, RequestOptions.DEFAULT)
    if(!isExists){
      return null
    }
    // 从es中提取制定的mid字段
    val mids: ListBuffer[String] = ListBuffer[String]()
    val searchRequest: SearchRequest = new SearchRequest(indexName)
    val searchSourceBuilder:SearchSourceBuilder = new SearchSourceBuilder()
    searchSourceBuilder.fetchSource(fieldName,null).size(100000)
    searchRequest.source(searchSourceBuilder)
    val searchResponse: SearchResponse = esClient.search(searchRequest, RequestOptions.DEFAULT)
    val hits: Array[SearchHit] = searchResponse.getHits.getHits
    for (hit <- hits) {
      val sourceAsMap: util.Map[String, AnyRef] = hit.getSourceAsMap
      val mid: String = sourceAsMap.get(fieldName).toString
      mids.append(mid)
    }
    mids.toList
  }

}
